package com.junbao.message.infrastructure.core.dispatcher;

import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.junbao.commom.context.ClassPathXmlApplicationContext;
import com.junbao.message.application.service.MessageService;
import com.junbao.message.domain.model.NotifyMessage;
import com.junbao.message.infrastructure.core.AbstractDisruptorCore;
import com.junbao.message.infrastructure.core.DataConsumer;
import com.junbao.message.infrastructure.core.DataProducer;
import com.junbao.message.infrastructure.core.RequestDto;
import com.junbao.message.infrastructure.core.handler.MessageRetrySendHandler;
import com.junbao.message.infrastructure.core.producer.MessageRetrySendProducer;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.WorkProcessor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * 消息发送diruptor
 * @author yilin
 *
 * @param <T>
 */
@SuppressWarnings("all")
public class MessageRetrySendDisruptor extends AbstractDisruptorCore{

	private static final Logger LOGGER = LoggerFactory.getLogger(MessageRetrySendDisruptor.class);
	
	private static class DisruptorCoreInstance{
		private final static MessageRetrySendDisruptor INSTANCE = new MessageRetrySendDisruptor();
	}
	
	private MessageRetrySendDisruptor(){};
	
	public static MessageRetrySendDisruptor instance(){
		return DisruptorCoreInstance.INSTANCE;
	}
	
	/** 消费者数量*/
	private static final int CONSUMER_COUNT = (int) (Runtime.getRuntime().availableProcessors()/0.1);
	
	/** 最大线程数量*/
	private static final int MAX_THREAD_COUNT = (int) (Runtime.getRuntime().availableProcessors()/0.1);
	
	
	static{
		try {
			MessageRetrySendDisruptor disruptorCore = MessageRetrySendDisruptor.instance();
			//获取消费者
			DataConsumer[] messageConsumers = getmessageConsumers(disruptorCore.setCounsumerClassType(),CONSUMER_COUNT);
			//启动disruptor
			disruptorCore.start(ProducerType.MULTI,
					RINGBUFFER_SIZE,
					new BlockingWaitStrategy(),
					messageConsumers,
					RequestDto.class);
		} catch (Exception e) {
			LOGGER.error("启动MessageRetrySendDisruptor失败！",e);
		}
	}

	@Override
	public DataProducer createDataProducer(String producerName) {
		return new MessageRetrySendProducer(producerName, this.ringBuffer);
	}
	
	@Override
	public String setWorkThreadName() {
		return "MessageRetrySend-thread";
	}
	
	@Override
	public int setCoreThreadSize() {
		return MAX_THREAD_COUNT;
	}
	
	@Override
	public int setMaxThreadSize() {
		return MAX_THREAD_COUNT;
	}

	@Override
	public Class setCounsumerClassType() {
		return MessageRetrySendHandler.class;
	}
	
	@Override
	public void rejectedExecutionHandler(Runnable runnable, ThreadPoolExecutor executor) {
		executor.execute(runnable);
	}

	@Override
	public void disruptorExceptionHandler(Throwable throwable, long sequence, Object event) {
		RequestDto<NotifyMessage> requestDto =(RequestDto)event;
		LOGGER.error("消息重试失败，消息编号："+requestDto.getBusinessObject().getMessageNo(),throwable);
		MessageService messageService = (MessageService) ClassPathXmlApplicationContext.getInstance().getBean("messageService");
		messageService.retryMessage(requestDto.getBusinessObject());
	}
}
